package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/8/28 10:24
 */
public class Datagen2Print {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id       BIGINT,\n" +     //采集信息id
                "    sid      STRING,\n" +    //水位传感器id
                "    ts   TIMESTAMP,\n" +    //数据采集时间
//                "    ts3   TIMESTAMP(3),\n" +    //数据采集时间
//                "    t   TIME,\n" +    //数据采集时间
//                "    dt   DATE,\n" +    //数据采集时间
//                "    ts_ltz   TIMESTAMP_LTZ,\n" +    //数据采集时间
                "    flag   BOOLEAN,\n" +       //水位报警开关
                "    vc INT,\n" +                //采集到的水位高度
                "    testdouble DOUBLE,\n" +                //采集到的水位高度
                "    testdecimal DECIMAL(10, 2)\n" +                //采集到的水位高度
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +   //一次生成100万条数据
                "  'rows-per-second' = '1', \n" +   //每秒生成10000条数据
                "  'fields.sid.length' = '5', \n" +   //传感器id长度
                "  'fields.id.min' = '1', \n" +
                "  'fields.id.max' = '100', \n" +
                "  'fields.vc.min' = '10', \n" +
                "  'fields.vc.max' = '100'\n" +
                ")");

//        tableEnv.executeSql("CREATE TEMPORARY VIEW node6 as SELECT * , PROCTIME() as alarm_time FROM sourceTable");
//        tableEnv.executeSql("CREATE TEMPORARY VIEW node6 as select * from `sourceTable` where (vc NOT BETWEEN 1 AND 20) AND (id BETWEEN 22 AND 44)");
        tableEnv.executeSql("CREATE TEMPORARY VIEW node6 as select CURRENT_DATE, CURRENT_TIMESTAMP, LOCALTIMESTAMP, DATE_FORMAT(LOCALTIMESTAMP, 'yyyyMMdd')  from `sourceTable` where (id <> 0)");
//        tableEnv.executeSql("CREATE TEMPORARY VIEW node6 as select " +
//                "ts, " +
//                "LOCALTIMESTAMP, " +
//                "TO_TIMESTAMP('2023-09-06 07:34:30.763')  " +
//                "DATE_FORMAT(TO_TIMESTAMP('2023-09-06 07:34:30.763'), 'yyyyMMdd')  " +
////                "DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:ss.SSS', 'yyyyMMdd') " +
////                "TO_TIMESTAMP(ts)  " +
//                "from `sourceTable` where (id <> 0)");

        Table table = tableEnv.from("node6");
        tableEnv.toDataStream(table).print();

        env.execute();


//        tableEnv.executeSql("CREATE TABLE sinkTable (" +
//                "id BIGINT, \n" +
//                "sid STRING, \n" +
//                "ts TIMESTAMP(3), \n" +
//                "flag BOOLEAN, \n" +
//                "proc_time TIMESTAMP_LTZ(3), \n" +
//                "vc INT) \n" +
//                " WITH ('connector' = 'print')\n");
////                + "LIKE sourceTable (EXCLUDING ALL)");
//
//        //Insert
//        tableEnv.executeSql("insert into sinkTable select * from sourceTable where c!=");
    }

}
